from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.window import Window
from pyspark.sql.functions import max as pyspark_max  # 导入聚合函数
# 创建SparkSession
spark = SparkSession.builder \
    .appName("义乌小商品店铺人气分析") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()


# 读取数据
def read_data(file_path):
    df = spark.read.csv(file_path, header=True, inferSchema=True)
    print("数据基本信息：")
    df.printSchema()
    print(f"数据总行数: {df.count()}")
    return df


# 数据预处理
def preprocess_data(df):
    # 将交易时间转换为日期和小时
    df = df.withColumn("交易日期", to_date("交易时间"))
    df = df.withColumn("交易小时", hour("交易时间"))

    # 添加星期几信息 (1-7, 1代表星期一)
    df = df.withColumn("星期几", dayofweek("交易日期"))

    # 添加月份信息
    df = df.withColumn("月份", month("交易日期"))

    return df


def calculate_shop_popularity(df):
    # 按店铺和日期分组，计算每日指标
    daily_metrics = df.groupBy("店铺名称", "交易日期") \
        .agg(
        sum("销售额").alias("日销售额"),
        sum("客流量").alias("日客流量"),
        avg("客单价").alias("日平均客单价"),
        count("交易ID").alias("日交易次数")
    )

    # 计算店铺总指标
    shop_metrics = daily_metrics.groupBy("店铺名称") \
        .agg(
        avg("日销售额").alias("平均日销售额"),
        avg("日客流量").alias("平均日客流量"),
        avg("日平均客单价").alias("平均客单价"),
        avg("日交易次数").alias("平均日交易次数"),
        sum("日客流量").alias("总客流量"),
        sum("日销售额").alias("总销售额")
    )

    # 计算各列的最大值
    max_avg_flow = shop_metrics.select(pyspark_max("平均日客流量")).collect()[0][0]
    max_avg_sales = shop_metrics.select(pyspark_max("平均日销售额")).collect()[0][0]
    max_avg_price = shop_metrics.select(pyspark_max("平均客单价")).collect()[0][0]

    # 计算人气评分 (综合考虑客流量、销售额和客单价)
    shop_metrics = shop_metrics.withColumn(
        "人气评分",
        round((col("平均日客流量") / max_avg_flow * 0.4 +
               col("平均日销售额") / max_avg_sales * 0.4 +
               col("平均客单价") / max_avg_price * 0.2) * 100, 2)
    )

    # 按人气评分排序
    shop_metrics = shop_metrics.orderBy(col("人气评分").desc())

    return shop_metrics


# 分析商品类别与店铺人气关系
def analyze_category_popularity(df):
    # 按商品类别和店铺分组
    category_shop = df.groupBy("商品类别", "店铺名称") \
        .agg(
        sum("销售额").alias("类别销售额"),
        sum("客流量").alias("类别客流量"),
        count("交易ID").alias("类别交易次数")
    )

    # 找出每个商品类别下最受欢迎的店铺
    window_spec = Window.partitionBy("商品类别").orderBy(col("类别客流量").desc())
    top_shops_by_category = category_shop.withColumn("排名", rank().over(window_spec)) \
        .filter(col("排名") <= 3)

    return top_shops_by_category


# 时间趋势分析
def analyze_time_trends(df):
    # 按日期分析
    daily_trends = df.groupBy("交易日期") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ) \
        .orderBy("交易日期")

    # 按星期分析
    weekday_trends = df.groupBy("星期几") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ) \
        .orderBy("星期几")

    # 按小时分析
    hourly_trends = df.groupBy("交易小时") \
        .agg(
        sum("销售额").alias("总销售额"),
        sum("客流量").alias("总客流量"),
        count("交易ID").alias("总交易次数")
    ) \
        .orderBy("交易小时")

    return daily_trends, weekday_trends, hourly_trends


# 主函数
def main():
    print("===== 开始义乌小商品店铺人气分析 =====")

    # 读取数据
    sales_data = read_data("sales_data.csv")

    # 数据预处理
    processed_data = preprocess_data(sales_data)

    # 计算店铺人气
    shop_popularity = calculate_shop_popularity(processed_data)
    print("\n===== 店铺人气排名 =====")
    shop_popularity.show(10)

    # 分析商品类别与店铺人气关系
    category_popularity = analyze_category_popularity(processed_data)
    print("\n===== 各类别最受欢迎的店铺 =====")
    category_popularity.show(30)

    # 时间趋势分析
    daily_trends, weekday_trends, hourly_trends = analyze_time_trends(processed_data)
    print("\n===== 按日期的销售趋势 =====")
    daily_trends.show(10)
    print("\n===== 按星期的销售趋势 =====")
    weekday_trends.show()
    print("\n===== 按小时的销售趋势 =====")
    hourly_trends.show()

    # 将结果保存为CSV文件
    shop_popularity.write.csv("shop_popularity_results.csv", header=True, mode="overwrite")
    category_popularity.write.csv("category_popularity_results.csv", header=True, mode="overwrite")

    print("\n===== 分析完成，结果已保存 =====")


if __name__ == "__main__":
    main()
    # 停止SparkSession
    spark.stop()